package coordinator

import (
	//"bytes"
	customrpc "distributed-database/rpcPackage"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/rpc"
	"strconv"
	"strings"
	"time"
)

const (
	node0port = "50051"
	node1port = "50052"
	node2port = "50053"
)
var (
	resultCh = make(chan []byte, 8) // 简化实现：返回最近一次处理结果
)
type RequestHandler struct {
	serverAddr []string
	client0    *rpc.Client
	client1    *rpc.Client
	client2    *rpc.Client
	timeout    time.Duration
}

const (
	UserTable      = "user"
	VideoTable     = "video"
	FollowTable    = "follow"
	LikeRecord     = "like_record"
	Favorite       = "favorite"
	CommentTable   = "comment"
	TagTable       = "tag"
	VideoTagTable  = "video_tag"
	ActionLogTable = "action_log"

	OperationInsert = "insert"
	OperationUpdate = "update"
	OperationDelete = "delete"
	OperationQuery  = "query"
)

//func StartRPCServer()

// 把json字符串变成请求结构体
func (h *RequestHandler) ParseRequest(jsonStr string) (*customrpc.DBRequest, error) {
	req := customrpc.DBRequest{}
	err := json.Unmarshal([]byte(jsonStr), &req)
	if err != nil {
		return nil, fmt.Errorf("解析请求失败：%v", err)
	}
	validOperations := map[string]bool{
		OperationInsert: true,
		OperationUpdate: true,
		OperationDelete: true,
		OperationQuery:  true,
	}
	if !validOperations[req.Operation] {
		return nil, fmt.Errorf("不支持的操作类型:%s", req.Operation)
	}
	return &req, nil
}
func (h *RequestHandler) ReturnToFront(reply *customrpc.CoordinatorReply) error {
	if reply == nil {
		return fmt.Errorf("reply 为空")
	}

	// 创建一个标准响应结构
	response := struct {
		InsertSuccess bool        `json:"insert_success"`
		QuerySuccess bool        `json:"query_success"`
		DeleteSuccess bool        `json:"delete_success"`
		Data    any `json:"data,omitempty"`
		Error   string      `json:"error,omitempty"`
	}{
		InsertSuccess: reply.InsertSuccess,
		QuerySuccess: reply.QuerySuccess,
		DeleteSuccess: reply.DeleteSuccess,
		Data:    reply.Data,
	}

	// 根据数据类型进行特殊处理
	if reply.Data != nil {
		switch data := reply.Data.(type) {
		case customrpc.User:
			response.Data = data
		case *customrpc.User:
			response.Data = *data
		case []customrpc.User:
			response.Data = data
		case []*customrpc.User:
			users := make([]customrpc.User, len(data))
			for i, user := range data {
				users[i] = *user
			}
			response.Data = users
		case customrpc.Video:
			response.Data = data
		case *customrpc.Video:
			response.Data = *data
		case []customrpc.Video:
			response.Data = data
		case []*customrpc.Video:
			videos := make([]customrpc.Video, len(data))
			for i, video := range data {
				videos[i] = *video
			}
			response.Data = videos
		// 添加其他类型的处理...
		default:
			// 其他类型保持原样
		}
	}

	// 转换为 JSON 字符串
	jsonData, err := json.Marshal(response)
	if err != nil {
		return fmt.Errorf("转换JSON失败: %v", err)
	}

	// 将 JSON 结果放入 resultCh，由 /api/request 的处理函数直接响应给前端
	select {
	case resultCh <- jsonData:
		fmt.Printf("结果已投递到 resultCh，长度=%d\n", len(jsonData))
		fmt.Printf("结果内容：%s\n", string(jsonData))
	default:
		// 若通道已满则丢弃最旧一条，保证最新结果可返回
		select {
		case <-resultCh:
		default:
		}
		resultCh <- jsonData
	}
	return nil
}
func (h *RequestHandler) GetShardKey(db *customrpc.DBRequest) int {
	// 安全的整型转换
	toInt := func(v interface{}) int {
		switch t := v.(type) {
		case int:
			return t
		case int64:
			return int(t)
		case float64:
			return int(t)
		case json.Number:
			if i64, err := t.Int64(); err == nil {
				return int(i64)
			}
			return 0
		case string:
			if t == "" {
				return 0
			}
			if i, err := strconv.Atoi(t); err == nil {
				return i
			}
			return 0
		default:
			return 0
		}
	}
	var key int
	// 根据表名和字段选取分片键值（按业务约定）
	switch db.Table {
	case UserTable:
		key = toInt(db.Data["user_id"]) // user 按 user_id 分片
	case VideoTable:
		key = toInt(db.Data["user_id"]) // video 按上传者 user_id 分片
	case FollowTable:
		key = toInt(db.Data["follower_id"]) // follow 按 follower_id 分片
	case LikeRecord:
		key = toInt(db.Data["user_id"]) // like_record 按 user_id 分片
	case Favorite:
		key = toInt(db.Data["user_id"]) // favorite 按 user_id 分片
	case CommentTable:
		key = toInt(db.Data["video_id"]) // comment 按 video_id 分片
	case TagTable:
		key = toInt(db.Data["tag_id"]) // tag 小表，按 tag_id 取模（或可固定0）
	case VideoTagTable:
		key = toInt(db.Data["video_id"]) // video_tag 按 video_id 分片
	case ActionLogTable:
		key = toInt(db.Data["user_id"]) // action_log 按 user_id 分片
	default:
		fmt.Printf("未知的数据表：%s，无法获取分片键值\n", db.Table)
		key = 0
	}
	return key % customrpc.PieceNum
}

// getMap 将 DBRequest.Data 规整为 map[string]interface{}，并做类型修正
// 规则：
// - 以 _id 结尾或常见主键字段，转为 int
// - 以 _count 结尾的计数字段，转为 int64
// - 时间字段（*_time、register_time、last_login、upload_time、action_time 等）统一解析为 "YYYY-MM-DD HH:MM:SS" 字符串
// - gender/duration 等常见数值字段做数值化
func (h *RequestHandler) getMap(d *customrpc.DBRequest) (map[string]interface{}, error) {
	if d == nil {
		return nil, fmt.Errorf("DBRequest 为空")
	}
	m := make(map[string]interface{}, len(d.Data))
	for k, v := range d.Data {
		m[k] = v
	}

	toInt := func(v interface{}) (int, bool) {
		switch t := v.(type) {
		case int:
			return t, true
		case int64:
			return int(t), true
		case float64:
			return int(t), true
		case json.Number:
			if i64, err := t.Int64(); err == nil {
				return int(i64), true
			}
			return 0, false
		case string:
			if t == "" {
				return 0, false
			}
			if i, err := strconv.Atoi(t); err == nil {
				return i, true
			}
			return 0, false
		default:
			return 0, false
		}
	}
	toInt64 := func(v interface{}) (int64, bool) {
		switch t := v.(type) {
		case int:
			return int64(t), true
		case int64:
			return t, true
		case float64:
			return int64(t), true
		case json.Number:
			if i64, err := t.Int64(); err == nil {
				return i64, true
			}
			return 0, false
		case string:
			if t == "" {
				return 0, false
			}
			if i64, err := strconv.ParseInt(t, 10, 64); err == nil {
				return i64, true
			}
			return 0, false
		default:
			return 0, false
		}
	}
	toTimeString := func(v interface{}) (string, bool) {
		switch t := v.(type) {
		case time.Time:
			return t.Format("2006-01-02 15:04:05"), true
		case string:
			if t == "" {
				return "", false
			}
			// RFC3339
			if tt, err := time.Parse(time.RFC3339, t); err == nil {
				return tt.Format("2006-01-02 15:04:05"), true
			}
			// 常见 SQL 格式
			if tt, err := time.Parse("2006-01-02 15:04:05", t); err == nil {
				return tt.Format("2006-01-02 15:04:05"), true
			}
			return "", false
		default:
			return "", false
		}
	}

	// 常见字段类别
	timeFields := map[string]struct{}{
		"register_time": {}, "last_login": {}, "upload_time": {},
		"follow_time": {}, "like_time": {}, "favorite_time": {},
		"comment_time": {}, "action_time": {},
	}
	intHint := map[string]struct{}{
		"user_id": {}, "video_id": {}, "follower_id": {}, "followee_id": {},
		"comment_id": {}, "tag_id": {}, "log_id": {}, "duration": {},
		// 某些表按 user_id 分片，这里只做数据规整
	}
	// gender 作为字符串字段，保持原值（"male" 或 "female"）

	for k, v := range m {
		lower := strings.ToLower(k)
		if _, ok := timeFields[lower]; ok || strings.HasSuffix(lower, "_time") {
			if s, ok := toTimeString(v); ok {
				m[k] = s
				continue
			}
		}
		if strings.HasSuffix(lower, "_count") {
			if i64, ok := toInt64(v); ok {
				m[k] = i64
			}
			continue
		}
		if strings.HasSuffix(lower, "_id") || (func() bool { _, ok := intHint[lower]; return ok })() {
			if i, ok := toInt(v); ok {
				m[k] = i
			}
			continue
		}
		// 其他字段保持原样
	}
	return m, nil
}
func (h *RequestHandler) HandleRequest(req *customrpc.DBRequest) error {
	// 1) 规整数据为 map 并做类型修正
	dataMap, err := h.getMap(req)
	if err != nil {
		return fmt.Errorf("规整请求数据失败: %v", err)
	}
	// 2) 计算分片键
	shard := h.GetShardKey(&customrpc.DBRequest{Table: req.Table, Data: dataMap})
	fmt.Printf("现在要处理的这个请求的键值是%d\n", shard)

	// 3) 组装 RPC 参数
	args := customrpc.CoordinatorRequest{
		TargetId:  shard,
		Data:      dataMap,
		Operation: req.Operation,
		Condition: req.Condition,
		TableName: req.Table,
	}
	go func() {
		reply := &customrpc.CoordinatorReply{}
		err := h.client0.Call("node.RecvRequestToNode", args, reply)
		if err != nil {
			fmt.Printf("节点0处理请求失败: %v\n", err)
			return
		}
		if reply.HandleOk {
			fmt.Printf("该请求被节点0处理\n")
			h.ReturnToFront(reply)
		} else {
			fmt.Printf("该请求未被节点0处理\n")
		}
	}()
	go func() {
		reply := &customrpc.CoordinatorReply{}
		h.client1.Call("node.RecvRequestToNode", args, reply)
		if reply.HandleOk {
			fmt.Printf("该请求被节点1处理\n")
			h.ReturnToFront(reply)
			return
		} else {
			fmt.Printf("该请求未被节点1处理\n")
			return
		}
	}()
	go func() {
		reply := &customrpc.CoordinatorReply{}
		err := h.client2.Call("node.RecvRequestToNode", args, reply)
		if err != nil {
			fmt.Printf("节点2处理请求失败: %v\n", err)
			return
		}
		if reply.HandleOk {
			fmt.Printf("该请求被节点2处理\n")
			h.ReturnToFront(reply)
			return
		} else {
			fmt.Printf("该请求未被节点2处理\n")
			return
		}
	}()
	return nil
}

// 根据键值选出目标节点
// 这个是启动rpc客户端，连接各个节点的rpc服务器
func (h *RequestHandler) StartrpcServer() error {
	node0address := "localhost:" + node0port
	node1address := "localhost:" + node1port
	node2address := "localhost:" + node2port
	//连接各个节点的rpc服务器
	var err error

	// 连接节点0
	client0, err := rpc.Dial("tcp", node0address)
	if err != nil {
		return fmt.Errorf("连接节点0RPC服务器失败(%s:%v)", node0address, err)
	}
	h.client0 = client0

	// 连接节点1
	client1, err := rpc.Dial("tcp", node1address)
	if err != nil {
		h.client0.Close() // 清理已建立的连接
		return fmt.Errorf("连接节点1RPC服务器失败(%s:%v)", node1address, err)
	}
	h.client1 = client1

	// 连接节点2
	client2, err := rpc.Dial("tcp", node2address)
	if err != nil {
		h.client0.Close() // 清理已建立的连接
		h.client1.Close()
		return fmt.Errorf("连接节点2RPC服务器失败(%s:%v)", node2address, err)
	}
	h.client2 = client2
	fmt.Printf("成功连接到所有RPC服务器\n")
	return nil
}
func (h *RequestHandler) Close() error {
	// 关闭所有客户端连接
	if h.client0 != nil {
		h.client0.Close()
	}
	if h.client1 != nil {
		h.client1.Close()
	}
	if h.client2 != nil {
		h.client2.Close()
	}
	return nil
}

type NodeInfo struct { //节点元数据
	NodeID        string
	Address       string
	Status        string
	Load          int //节点当前负载，其实感觉也没多大用
	LastHeartbeat time.Time
}

type Coordinator struct {
}

// StartHTTPServer 启动 HTTP 服务器来接收前端数据
// StartHTTPServer 启动 HTTP 服务器来接收前端数据
func StartHTTPServer(port int, dataChan chan string) {
	// 简单 CORS 处理
	withCORS := func(w http.ResponseWriter) {
		w.Header().Set("Access-Control-Allow-Origin", "*")
		w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
		w.Header().Set("Access-Control-Allow-Methods", "POST, OPTIONS")
	}

	http.HandleFunc("/api/request", func(w http.ResponseWriter, r *http.Request) {
		withCORS(w)
		if r.Method == http.MethodOptions {
			w.WriteHeader(http.StatusNoContent)
			return
		}
		if r.Method != http.MethodPost {
			http.Error(w, "只允许 POST 方法", http.StatusMethodNotAllowed)
			return
		}

		// 读取请求体
		body, err := io.ReadAll(r.Body)
		if err != nil {
			http.Error(w, "读取请求体失败", http.StatusInternalServerError)
			log.Printf("读取请求体错误: %v\n", err)
			return
		}
		defer r.Body.Close()

		// 将数据发送到通道进行异步处理
		dataChan <- string(body)
		// 等待处理完成后由 ReturnToFront 放入 resultCh，再回给前端
		select {
		case res := <-resultCh:
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusOK)
			w.Write(res)
		case <-time.After(10 * time.Second):
			// 超时也返回 200，让前端拿到业务失败信息而不是抛出网络错误
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusOK)
			json.NewEncoder(w).Encode(map[string]any{
				"success": false,
				"error":   "处理超时或无可用节点处理",
			})
		}
	})

	addr := fmt.Sprintf(":%d", port)
	fmt.Printf("HTTP 服务器正在监听端口 %d...\n", port)
	if err := http.ListenAndServe(addr, nil); err != nil {
		fmt.Printf("HTTP 服务器启动失败: %v\n", err)
	}
}

// // 整个项目的入口处
// func main() {
// 	fmt.Printf("coordinator开启，注册RPC类型...")
// 	// 注册所有要通过RPC传输的类型
// 	customrpc.RegisterTypes()

// 	fmt.Printf("开始创建数据通道")
// 	// 创建数据通道
// 	dataChan := make(chan string, 100)
// 	fmt.Printf("启动http服务器，监听8080端口")
// 	// 启动 HTTP 服务器
// 	go StartHTTPServer(8080, dataChan)
// 	fmt.Printf("连接rpc服务器...")
// 	// 初始化请求处理器
// 	rh := &RequestHandler{}
// 	err := rh.StartrpcServer()
// 	if err != nil {
// 		log.Fatalf("连接RPC服务器错误: %v", err)
// 	}
// 	defer rh.Close()
// 	fmt.Printf("rpc服务器连接成功！")
// 	// 主循环处理接收到的数据
// 	// for receiveString := range dataChan {
// 	receiveString := `{
//         "request_id": "req-insert-1",
//         "operation": "insert",
//         "database": "short_video",
//         "table": "user",
//         "data": {
//             "user_id": "123",
//             "username": "alice",
//             "password_hash": "hash",
//             "nickname": "Alice",
//             "avatar_url": "http://example.com/a.jpg",
//             "gender": "1",
//             "bio": "hello",
//             "phone": "1234567890",
//             "email": "alice@example.com",
//             "register_time": "2025-10-25T12:00:00Z",
//             "last_login": "2025-10-25T12:00:00Z",
//             "status": "active"
//         }
//     }`
// 	db, err := rh.ParseRequest(receiveString) //原始字符串转换成结构体
// 	if err != nil {
// 		log.Printf("原始请求字符串解析为请求结构体失败: %v", err)
// 		return
// 		//continue
// 	}

// 	if err := rh.HandleRequest(db); err != nil { //结构体转换成数据表
// 		log.Printf("处理请求失败: %v", err)
// 		return
// 		//continue
// 	}

// 	log.Printf("成功处理请求")
// 	// }

// }
